package com.lagou.dw.flume.interceptor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.compress.utils.Charsets;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.interceptor.Interceptor;
import org.junit.Test;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;

public class LogTypeInterceptor implements Interceptor {

    @Override
    public void initialize() {

    }

    // 逐条处理event
    @Override
    public Event intercept(Event event) {
        // 获取 event 的 body
        String eventBody = new String(event.getBody(), Charsets.UTF_8);
        // 获取 event 的 header
        Map<java.lang.String, java.lang.String> headersMap = event.getHeaders();
        // 解析body获取json串
        String[] bodyArr = eventBody.split("\\s+");

        try {
            //String jsonStr = bodyArr[6];
            //可能json会有空格，截取6之后的搜索数组组成json字符串
            String[] newBodyArr = Arrays.copyOfRange(bodyArr, 6, bodyArr.length);
            String jsonStr = StringUtils.join(newBodyArr);
            // 解析json串获取时间戳
            String timestampStr = "";
            //System.out.println(jsonStr);
            JSONObject jsonObject = JSON.parseObject(jsonStr);
            if(headersMap.getOrDefault("logtype","").equals("start")){
                // 取启动日志的时间戳
                timestampStr = jsonObject.getJSONObject("app_active").getString("time");
            }else if(headersMap.getOrDefault("logtype","").equals("event")){
                // 取事件日志第一条记录的时间戳
                JSONArray jsonArray = jsonObject.getJSONArray("lagou_event");
                if(jsonArray.size()>0){
                    timestampStr = jsonArray.getJSONObject(0).getString("time");
                }
            }

            // 将时间戳转换为字符串 "yyyy-MM-dd"
            // 将字符串转换为Long
            long timestamp = Long.parseLong(timestampStr);
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
            Instant instant = Instant.ofEpochMilli(timestamp);
            LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
            String date = formatter.format(localDateTime);
            // 将转换后的字符串放置header中
            headersMap.put("logtime",date);
            event.setHeaders(headersMap);
        } catch (Exception e) {
            //System.out.println(e.getMessage());
            headersMap.put("logtime","unknow");
            event.setHeaders(headersMap);
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        ArrayList<Event> listEvent = new ArrayList<>();
        for (Event event:events){
            Event outEvent = intercept(event);
            if( outEvent != null ) {
                listEvent.add(outEvent);
            }
        }
        return listEvent;
    }

    public static class Builder implements Interceptor.Builder{

        @Override
        public Interceptor build() {
            return new LogTypeInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }

    @Override
    public void close() {

    }

    @Test
    public void startJunit(){
        String str = "2021-09-16 16:55:01.203 [main] INFO  com.lagou.ecommerce.AppStart - {\"app_active\":{\"name\":\"app_active\",\"json\":{\"entry\":\"1\",\"action\":\"0\",\"error_code\":\"0\"},\"time\":1595260800000},\"attr\":{\"area\":\"连云港\",\"uid\":\"2F10092A1\",\"app_v\":\"1.1.8\",\"event_type\":\"common\",\"device_id\":\"1FB872-9A1001\",\"os_type\":\"0.43\",\"channel\":\"PN\",\"language\":\"chinese\",\"brand\":\"iphone-7\"}}";
        HashMap<String, String> map = new HashMap<>();
        // new Event
        SimpleEvent event = new SimpleEvent();
        map.put("logtype","start");
        event.setHeaders(map);
        event.setBody(str.getBytes(Charsets.UTF_8));
        // 调用interceptor处理event
        LogTypeInterceptor logTypeInterceptor = new LogTypeInterceptor();
        Event outEvent = logTypeInterceptor.intercept(event);
        // 处理结果
        Map<String, String> headersMap = outEvent.getHeaders();
        System.out.println(JSON.toJSONString(headersMap));
    }

    @Test
    public void eventJunit(){
        String str = "2021-09-16 16:54:41.480 [main] INFO  com.lagou.ecommerce.AppEvent - {\"lagou_event\":[{\"name\":\"goods_detail_loading\",\"json\":{\"entry\":\"3\",\"goodsid\":\"0\",\"loading_time\":\"12\",\"action\":\"1\",\"staytime\":\"38\",\"showtype\":\"2\"},\"time\":1595260800000},{\"name\":\"loading\",\"json\":{\"loading_time\":\"21\",\"action\":\"1\",\"loading_type\":\"1\",\"type\":\"1\"},\"time\":1595260800000},{\"name\":\"ad\",\"json\":{\"duration\":\"26\",\"ad_action\":\"0\",\"shop_id\":\"0\",\"event_type\":\"ad\",\"ad_type\":\"4\",\"show_style\":\"0\",\"product_id\":\"9\",\"place\":\"placecampaign2_index\",\"sort\":\"8\"},\"time\":1595260800000},{\"name\":\"comment\",\"json\":{\"praise_count\":447,\"comment_id\":2,\"reply_count\":25,\"userid\":4,\"content\":\"Non voluptatem reprehenderit voluptatem velit minus non ut sunt et sed amet nemo vel sunt rem libero deserunt aliquam vitae dicta ut aut dolores molestiae fuga excepturi est occaecati.\"},\"time\":1595260800000},{\"name\":\"favorites\",\"json\":{\"course_id\":2,\"id\":0,\"userid\":0},\"time\":1595260800000}],\"attr\":{\"area\":\"扬州\",\"uid\":\"2F10092A0\",\"app_v\":\"1.1.6\",\"event_type\":\"common\",\"device_id\":\"1FB872-9A1000\",\"os_type\":\"4.71\",\"channel\":\"MJ\",\"language\":\"chinese\",\"brand\":\"Huawei-8\"}}";
        HashMap<String, String> map = new HashMap<>();
        // new Event
        SimpleEvent event = new SimpleEvent();
        map.put("logtype","event");
        event.setHeaders(map);
        event.setBody(str.getBytes(Charsets.UTF_8));
        // 调用interceptor处理event
        LogTypeInterceptor logTypeInterceptor = new LogTypeInterceptor();
        Event outEvent = logTypeInterceptor.intercept(event);
        // 处理结果
        Map<String, String> headersMap = outEvent.getHeaders();
        System.out.println(JSON.toJSONString(headersMap));
    }
}
